package com.bxl.chooser.client;

import com.bxl.chooser.bean.AppInfo;
import com.bxl.chooser.bean.ChooseMsg;
import com.bxl.chooser.bean.ChooseResponseMsg;
import com.bxl.chooser.bean.MessageBean;
import com.bxl.chooser.context.ChooserContext;
import com.bxl.chooser.factory.ChooseMsgFactory;
import com.bxl.chooser.util.MessageRegister;
import com.bxl.chooser.util.ThreadUtil;
import com.bxl.netty.rpc.client.common.client.ClientInfo;
import com.bxl.netty.rpc.common.future.CompletableFuture;
import com.bxl.netty.rpc.common.message.PackageMsg;
import io.netty.channel.socket.nio.NioSocketChannel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * leader需要维护每个Partner的 logindex 进行 数据的同步
 *
 * @author baixl
 * @date 2021/3/25
 */
public class Partners {
    private Logger logger = LoggerFactory.getLogger(Partners.class);
    private List<ClientInfo> clients;
    private ChooserContext context;
    private AtomicInteger voteNum = new AtomicInteger();

    private List<ClientLogRecord> noSyncList = new ArrayList<>();
    private List<ClientInfo> syncList = new ArrayList<>();
    private CommitTask commitTask;

    //leader模拟断网
    private int index = 0;

    public int getPartnersNum() {
        return clients.size();
    }

    public Partners(ChooserContext context) {
        this.context = context;
        init();
    }

    private void init() {
        clients = new ArrayList<>();
        clients.add(new ClientInfo("part", "localhost", 9090, context.getClientContext()));
        clients.add(new ClientInfo("part", "localhost", 9091, context.getClientContext()));
        clients.add(new ClientInfo("part", "localhost", 9092, context.getClientContext()));
        syncList.addAll(clients);
        //clients.add(new ClientInfo("part", "localhost", 9093, context.getClientContext()));
        //clients.add(new ClientInfo("part", "localhost", 9094, context.getClientContext()));
    }

    public int vote() {
        return sendMsg(1, "", "", -1);
    }

    public void stopSync() {
        noSyncList.forEach(item -> item.setStop(true));
    }

    public void startCommit(AppInfo appInfo) {
        if (appInfo.getState() == AppInfo.AppState.LEADER) {
            this.commitTask = new CommitTask(this, appInfo);
            logger.info("开启数据提交");
            Thread t = new Thread(this.commitTask);
            t.setDaemon(true);
            t.start();
        }
    }

    public void stopCommit() {
        logger.info("停止数据提交");
        this.commitTask.setStop(true);
    }

    private static class CommitTask implements Runnable {
        private Partners partners;
        private AppInfo appInfo;
        private volatile boolean isStop = false;

        public CommitTask(Partners partners, AppInfo appInfo) {
            this.partners = partners;
            this.appInfo = appInfo;
        }

        public boolean isStop() {
            return isStop;
        }

        public void setStop(boolean stop) {
            isStop = stop;
        }

        @Override
        public void run() {
            while (!isStop && appInfo.getState() == AppInfo.AppState.LEADER) {
                if ((appInfo.getCommitLogIndex() + 1) >= appInfo.getLogIndex()) {
                    try {
                        TimeUnit.SECONDS.sleep(1);
                    } catch (InterruptedException e) {

                    }
                } else {
                    appInfo.commitLogIndexIncrease();
                    System.out.println(String.format("提交数据 %s", appInfo.getCommitLogIndex()));
                    int commit = partners.commit(appInfo.getCommitLogIndex());
                    //全部提交成功
                    if (commit >= (partners.getPartnersNum() / 2)) {
                        System.out.println(String.format("数据 %s 提交成功", appInfo.getCommitLogIndex()));
                    } else {
                        System.out.println(String.format("数据 %s 提交失败", appInfo.getCommitLogIndex()));
                    }
                }
            }
        }
    }


    private static class SyncTask implements Runnable {
        private ClientLogRecord clientLogRecord;
        private AppInfo appInfo;
        private Partners partners;

        public SyncTask(ClientLogRecord clientLogRecord, AppInfo appInfo, Partners partners) {
            this.clientLogRecord = clientLogRecord;
            this.appInfo = appInfo;
            this.partners = partners;
        }

        @Override
        public void run() {
            while (!clientLogRecord.isStop() && appInfo.getState() == AppInfo.AppState.LEADER) {
                System.out.println("数据同步");
                MessageBean msg = appInfo.getMsg(clientLogRecord.getIndex());
                if (msg != null) {
                    PackageMsg<ChooseMsg> registerMsg = ChooseMsgFactory.createMsg(5, appInfo, msg, -1);
                    ChooseResponseMsg chooseResponseMsg = partners.sendMsg(clientLogRecord.getClientInfo(), registerMsg);
                    if (chooseResponseMsg != null && chooseResponseMsg.isVote()) {
                        //写消息成功
                        clientLogRecord.indexIncrease();
                    }else if(chooseResponseMsg != null && !chooseResponseMsg.isVote()) {
                        //同步失败，自动调整logIndex
                        clientLogRecord.setIndex(chooseResponseMsg.getLogIndex());
                    }
                } else {
                    synchronized (partners.getSyncList()) {
                        //已经是同步数据
                        partners.getSyncList().add(clientLogRecord.getClientInfo());

                    }
                    synchronized (partners.getNoSyncList()) {
                        //已经是同步数据
                        partners.getNoSyncList().remove(clientLogRecord);

                    }
                    break;
                }

            }
        }
    }


    public int add(MessageBean messageBean) {
        voteNum = new AtomicInteger(0);
        int noSyncNum = noSyncList.size();
        syncList.parallelStream().forEach(node -> {
            if (node.getIp().equals(context.getIp()) && node.getPort() == context.getPort()) {
                //自己的一票
                voteNum.incrementAndGet();
            } else {
                PackageMsg<ChooseMsg> registerMsg = ChooseMsgFactory.createMsg(3, context.getAppInfo(), messageBean, -1);
                ChooseResponseMsg chooseResponseMsg = sendMsg(node, registerMsg);
                if (chooseResponseMsg != null && chooseResponseMsg.isVote()) {
                    //写消息成功
                    voteNum.incrementAndGet();
                } else {
                    //写消息失败,数据不同步导致，进行数据同步
                    if (chooseResponseMsg != null) {
                        if (chooseResponseMsg.getLogIndex() < context.getAppInfo().getLogIndex()) {
                            synchronized (noSyncList) {
                                ClientLogRecord clientLogRecord = new ClientLogRecord(node, chooseResponseMsg.getLogIndex());
                                ThreadUtil.submit(new SyncTask(clientLogRecord, context.getAppInfo(), this));
                                noSyncList.add(clientLogRecord);
                            }
                        }
                    }


                }
            }

        });
        if (noSyncNum != syncList.size()) {
            synchronized (syncList) {
                noSyncList.forEach(item -> {
                    syncList.remove(item.getClientInfo());
                });
            }
        }
        //加上自己的一票
        return voteNum.get();
    }

    public int commit(int commitIndex) {
        return sendMsg(4, "", "", commitIndex);
    }

    private ChooseResponseMsg sendMsg(ClientInfo server, PackageMsg<ChooseMsg> registerMsg) {
        ChooseResponseMsg responseMsg = null;
        NioSocketChannel client = server.getClient();
        CompletableFuture<ChooseResponseMsg> cf = new CompletableFuture<ChooseResponseMsg>();
        MessageRegister.register(registerMsg.getHeader().getRequestID(), cf);
        try {
            client.writeAndFlush(registerMsg);
            responseMsg = cf.get();
        } catch (Exception e) {
            e.printStackTrace();
        }
        return responseMsg;
    }


    public int sendMsg(int flag, String messageId, String content, int commitIndex) {
        AtomicInteger tempVoteNum = new AtomicInteger(0);
        clients.parallelStream().forEach(node -> {
            if (node.getIp().equals(context.getIp()) && node.getPort() == context.getPort()) {
                //自己的一票
                tempVoteNum.incrementAndGet();
            } else {
                PackageMsg<ChooseMsg> registerMsg = ChooseMsgFactory.createMsg(flag, context.getAppInfo(), messageId, content, commitIndex);
                ChooseResponseMsg chooseResponseMsg = sendMsg(node, registerMsg);
                if (chooseResponseMsg != null && chooseResponseMsg.isVote()) {
                    //投给自己了
                    tempVoteNum.incrementAndGet();
                } else {
                    //没有投给自己，更新自己的term
                    if (chooseResponseMsg != null && chooseResponseMsg.getTerm() > context.getAppInfo().getTerm()) {
                        context.getAppInfo().setTerm(chooseResponseMsg.getTerm());
                    }

                }
            }

        });
        //加上自己的一票
        return tempVoteNum.get();
    }


    /**
     * 心跳定时：等待上一次心跳发送完毕，在发送下一次心跳。
     */
    public void heartBeat() {
        System.out.println("i am leader");
        logger.info("curLogindex {}; commitLogindex {}", context.getAppInfo().getLogIndex(), context.getAppInfo().getCommitLogIndex());
        //模拟心跳丢失，会切换leader， 正常应该删除
        /*if (index == 10) {
            try {
                TimeUnit.SECONDS.sleep(61);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
        index++;*/
        //==================================================

        if (context.getAppInfo().getState() != AppInfo.AppState.LEADER) {
            return;
        }
        //leader心跳
        int noSyncNum = syncList.size();
        clients.parallelStream().forEach(node -> {
            if (node.getIp().equals(context.getIp()) && node.getPort() == context.getPort()) {
                return;
            }
            PackageMsg<ChooseMsg> registerMsg = ChooseMsgFactory.createMsg(2, context.getAppInfo(), "", "", -1);
            registerMsg.getContent().setLeaderIp(context.getLeader().getIp());
            registerMsg.getContent().setPort(context.getPort());
            NioSocketChannel client = node.getClient();
            CompletableFuture<ChooseResponseMsg> cf = new CompletableFuture<ChooseResponseMsg>();
            MessageRegister.register(registerMsg.getHeader().getRequestID(), cf);
            try {
                client.writeAndFlush(registerMsg);
                ChooseResponseMsg responseMsg = cf.get();
                //拒绝心跳信息，降级为follower
                if (responseMsg != null && !responseMsg.isVote() && responseMsg.getTerm() > context.getAppInfo().getTerm()) {
                    context.degraded();
                }
                if (responseMsg != null && responseMsg.isVote() && responseMsg.getTerm() == context.getAppInfo().getTerm()
                        && (context.getAppInfo().getLogIndex() - responseMsg.getLogIndex()) > 2 && !noSyncList.contains(node)) {
                    //索引相差太多，开启同步
                    synchronized (noSyncList) {
                        ClientLogRecord clientLogRecord = new ClientLogRecord(node, responseMsg.getLogIndex());
                        ThreadUtil.submit(new SyncTask(clientLogRecord, context.getAppInfo(), this));
                        noSyncList.add(clientLogRecord);
                    }
                }
            } catch (Exception e) {
                e.printStackTrace();
            }

        });

        if (noSyncNum != syncList.size()) {
            synchronized (syncList) {
                noSyncList.forEach(item -> {
                    syncList.remove(item.getClientInfo());
                });
            }
        }
    }

    private static class ClientLogRecord {
        private ClientInfo clientInfo;
        private int index;
        private volatile boolean isStop;

        public ClientLogRecord(ClientInfo clientInfo, int index) {
            this.clientInfo = clientInfo;
            this.index = index;
            this.isStop = false;
        }

        public ClientInfo getClientInfo() {
            return clientInfo;
        }

        public void setClientInfo(ClientInfo clientInfo) {
            this.clientInfo = clientInfo;
        }

        public int getIndex() {
            return index;
        }

        public void indexIncrease() {
            index++;
        }

        public void setIndex(int index) {
            this.index = index;
        }

        public boolean isStop() {
            return isStop;
        }

        public void setStop(boolean stop) {
            isStop = stop;
        }
    }

    public List<ClientInfo> getSyncList() {
        return syncList;
    }

    public void setSyncList(List<ClientInfo> syncList) {
        this.syncList = syncList;
    }

    public List<ClientLogRecord> getNoSyncList() {
        return noSyncList;
    }

    public void setNoSyncList(List<ClientLogRecord> noSyncList) {
        this.noSyncList = noSyncList;
    }
}
